package hub

import (
	"github.com/gorilla/websocket"
	"time"
)

// Sub 消费者  相当于从数据库里取消息！fanout群发 direct、topic
func (m *Message) Sub() {
	ch, err := conn.Channel()
	if err != nil {
		return
	}
	err = ch.ExchangeDeclare(
		m.Exchange, // name
		m.Kind,     // type
		false,      // durable
		false,      // auto-deleted
		false,      // internal
		false,      // no-wait
		nil,        // arguments
	)
	// 声明队列
	q, err := ch.QueueDeclare(
		m.Queue, // name
		false,   // durable
		false,   // delete when unused
		true,    // exclusive
		false,   // no-wait
		nil,     // arguments
	)
	// 绑定队列
	err = ch.QueueBind(
		q.Name,     // queue name
		m.Routing,  // routing key
		m.Exchange, // exchange
		false,
		nil,
	)
	// 注册消费者
	msgs, err := ch.Consume(
		q.Name,   // queue
		m.Userid, // consumer
		true,     // auto-ack  因为是聊天消息，可以丢失
		false,    // exclusive
		false,    // no-local
		false,    // no-wait
		nil,      // args
	)
	c := m.Connect
	ticker := time.NewTicker(time.Second * 9)
	defer func() {
		ticker.Stop()
		_ = c.Ws.Close()
	}()
	for {
		select {
		case d, ok := <-msgs: // 客户端产生的消息
			if !ok {
				_ = c.writePump(websocket.CloseMessage, []byte{})
				return // 断开连接
			}
			if err := c.writePump(websocket.TextMessage, d.Body); err != nil {
				return
			}
		case d, ok := <-c.Send: // 系统产生的消息
			if !ok {
				_ = c.writePump(websocket.CloseMessage, []byte{})
				return
			}
			if err := c.writePump(websocket.TextMessage, d); err != nil {
				return
			}
		case <-ticker.C: // websocket连接监控消息，每隔9s ping 一次，放在客户端意外退出，服务端没有接受到消息。
			if err := c.writePump(websocket.PingMessage, []byte{}); err != nil {
				return // 当出现err时断开连接
			}
		}
	}
}

func (c *Connect) writePump(mt int, payload []byte) error {
	_ = c.Ws.SetWriteDeadline(time.Now().Add(time.Second * 10))
	return c.Ws.WriteMessage(mt, payload)
}
